package com.test

import org.apache.hadoop.hive.common.AcidMetaDataFile.DataFormat
import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.{DataFrame, Row, SparkSession}
import org.apache.spark.ml.recommendation.ALS
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types.{DoubleType, IntegerType, StructType}
import org.apache.spark.sql.functions._
import org.apache.spark.ml.evaluation.RegressionEvaluator
import org.apache.spark.mllib.evaluation.RankingMetrics
import org.apache.spark.sql.functions.{col, expr}

import java.sql.{Connection, DriverManager, ResultSet, Statement}
import java.util.Properties
object AlsFromMysql {
  // 创建 SparkSession
  val spark = SparkSession.builder().appName("als-demo").master("local[*]").getOrCreate()

  // 测试数据库连接
  def testmysql(): Unit = {
    val jdbcUrl = "jdbc:mysql://localhost:3306/hf-test"
    val username = "root"
    val password = "123456"

    var connection: Connection = null
    var statement: Statement = null
    var resultSet: ResultSet = null

    try {
      // 设置 JDBC 连接属性
      val connectionProps = new Properties()
      connectionProps.put("user", username)
      connectionProps.put("password", password)

      // 打开数据库连接
      connection = DriverManager.getConnection(jdbcUrl, connectionProps)
      println("成功连接到数据库")

      // 创建 Statement 对象并执行查询
      statement = connection.createStatement()
      resultSet = statement.executeQuery("SELECT 1")

      // 输出查询结果
      while (resultSet.next()) {
        val result = resultSet.getInt(1)
        println(s"查询结果：$result")
      }
    } catch {
      case e: Exception =>
        println(s"连接数据库出错：${e.getMessage}")
        e.printStackTrace()
    } finally {
      // 关闭资源
      if (resultSet != null) resultSet.close()
      if (statement != null) statement.close()
      if (connection != null) connection.close()
    }
  }
  // 修改mapUserIdAndGoodsRatings方法
  def mapUserIdAndGoodsRatingsPlus(): DataFrame = {

    // 从 MySQL 数据库读取数据
    val props = new Properties()
    props.setProperty("user", "root")
    props.setProperty("password", "123456")
    val jdbcUrl = "jdbc:mysql://localhost:3306/hf"
    val query = "(SELECT user_id as userId, good_id as goodId, CAST(action AS DOUBLE) as rating FROM sys_user_good_action) AS tmp"

    val rating = spark.read.jdbc(jdbcUrl, query, props)


//    val rddData: RDD[((Int, Int), Double)] = rating.rdd.map {
//      case Row(userId: Long, goodId: Long, rating: Double) =>
//        ((userId.toInt, goodId.toInt), rating)
//    }.reduceByKey(_+_)

    val rddData: RDD[((Int, Int), Double)] = rating.rdd.map {
      case Row(userId: Long, goodId: Long, rating: Double) =>
        ((userId.toInt, goodId.toInt), rating)
    }.reduceByKey((a, b) => if (a > b) a else b)



    // 展示合并后的结果
    rddData.take(10).foreach(println)
//    ((5421, 10741), 1.0)
//    ((5706, 10831), 2.0)
//    ((1011, 10163), 3.0)
//    ((1091, 10002), 2.0)
//    ((3268, 10485), 1.0)
//    ((5694, 10821), 1.0)
    println("------rddData------")

    val aggregatedRatingsRow = rddData.map {
      case ((userId, itemId), ratingSum) =>
        Row(userId, itemId, ratingSum)
    }

    // 定义数据结构
    val schema = new StructType()
      .add("userId", IntegerType)
      .add("itemId", IntegerType)
      .add("rating", DoubleType)

    // 将RDD转换为DataFrame
    val dataFrame1 = spark.createDataFrame(aggregatedRatingsRow, schema)

    dataFrame1
  }




  // 计算模型性能指标
  def evaluateModel(predictions: DataFrame): (Double, Double) = {
    // 创建回归评估器
    val evaluator = new RegressionEvaluator()
      .setLabelCol("rating") // 设置真实评分列
      .setPredictionCol("prediction") // 设置预测评分列

    // 计算 RMSE
    val rmse = evaluator.setMetricName("rmse").evaluate(predictions)

    // 计算 MAE
    val mae = evaluator.setMetricName("mae").evaluate(predictions)

    (rmse, mae) // 返回 RMSE 和 MAE
  }


  def train(epochs: Int, rank: Int, regParam: Double, trainingData: DataFrame, testData: DataFrame): (Double, Double)  = {
    val als = new ALS()
      .setMaxIter(epochs) // 迭代次数，用于最小二乘交替迭代的次数
      .setRank(rank) // 隐向量的维度
      .setRegParam(regParam) // 惩罚系数
      .setUserCol("userId") // user_id
      .setItemCol("itemId") // item_id
      .setRatingCol("rating") // 评分列

    // 训练模型
    val model = als.fit(trainingData)

    // 在测试集上评估模型

    val predictions = model.transform(testData)
      .withColumn("prediction", col("prediction").cast("double"))
      .na.drop(Seq("prediction"))

//    val rowCount = predictions.count()
//    println("predictions DataFrame的行数1: " + rowCount)
    predictions.show()
//    val rowCount2 = predictions.na.drop(Seq("prediction")).count()
//    println("predictions DataFrame的行数2: " + rowCount2)


    // 使用 evaluateModel 函数评估模型性能
    val (rmse, mae) = evaluateModel(predictions)

    val perUserActual = predictions.where("rating>=1.5")
      .groupBy("userId").agg(expr("collect_set(itemId) as goods"))
    perUserActual.cache()
    perUserActual.show()
    println("------perUserActual------")

    val perUserPredictions = predictions.orderBy(col("userId"),col("Prediction").desc)
      .groupBy("userId").agg(expr("collect_list(ItemId) as goods2"))
    perUserPredictions.cache()



    perUserPredictions.show()
    println("------perUserPredictions------")

    import spark.implicits._
    val perUserActualvPred = perUserActual.join(perUserPredictions,Seq("userId"))

    perUserActualvPred.cache()
    perUserActualvPred.show()
    println("------perUserActualvPred------")

//    val perUserActualvPred2 = perUserActualvPred
//      .map(row => (
//        row(1).asInstanceOf[Seq[Integer]].toArray,
//        row(2).asInstanceOf[Seq[Integer]].toArray.take(15)
//      ))
    val perUserActualvPred2 = perUserActualvPred
      .map(row => {
        val goodsArray = row.getAs[Seq[Integer]]("goods").toArray
        val goods2Array = row.getAs[Seq[Integer]]("goods2").toArray
        val goods2Take15 = if (goods2Array.length >= 50) goods2Array.take(50) else goods2Array
        (goodsArray, goods2Take15)
      })

    perUserActualvPred2.cache()
    perUserActualvPred2.show()

    // 计算准确率
    val userPrecision = perUserActualvPred2.map { case (actual, predicted) =>
      // 交集计算实际与预测的相同元素的个数
      val intersection = predicted.take(10).intersect(actual).length
      // 判断实际列表是不是为空，非空就进行计算
      val precision = if (actual.nonEmpty) intersection.toDouble / actual.length else 0.0
      precision
    }


    // 计算平均准确率
    val AverageAccuracyByAction = userPrecision.agg(avg("value")).first().getDouble(0)
    userPrecision.show()
    println("----userPrecision-----")


    val ranks = new RankingMetrics(perUserActualvPred2.rdd)

    val AverageAccuracy = ranks.meanAveragePrecision
    val TopAccuracy = ranks.precisionAt(10)
    // 7-4 查看Top准确率
    println("平均准确率: " + AverageAccuracy) //平均准确率:Toby: 0.3580740856864441;: 0.3460270484759316
    println("Top准确率: " + TopAccuracy) //Top准确率:Toby: 0.3794871794871796;: 0.2257154882154882
    println("自定义准确率: "+ AverageAccuracyByAction)

    (rmse, mae)
  }




  def main(args: Array[String]): Unit = {
    // 设置日志级别
    Logger.getLogger("org").setLevel(Level.ERROR)

    val rating = mapUserIdAndGoodsRatingsPlus()

    // 划分数据集为训练集和测试集（90%训练，10%测试）
    val Array(trainingData, testData) = rating.randomSplit(Array(0.8, 0.2))

    // 展示前5条评分记录
//    rating.show(20)
    // +------+------+------+
    // |userId|itemId|rating|
    // +------+------+------+
    // |  5421| 10741|   1.0|
    // |  5706| 10831|   2.0|
    // |  1011| 10163|   3.0|
    // |  1091| 10002|   2.0|
    // |  3268| 10485|   1.0|
    // |  5694| 10821|   1.0|
    // |  2598| 10332|   1.0|
    // |  1951| 10824|   1.0|

//    rating.printSchema()


//    val (rmse1, mae1) = train(5,3,0.1,trainingData,testData)
//    val (rmse2, mae2) = train(10,3,0.1,trainingData,testData)
//    val (rmse3, mae3) = train(10,6,0.1,trainingData,testData)
//    val (rmse4, mae4) = train(15,3,0.1,trainingData,testData)
//    val (rmse5, mae5) = train(15,6,0.1,trainingData,testData)
    val (rmse6, mae6) = train(10,9,0.1,trainingData,testData)


    // 打印结果
//    println(s"轮次 = 5,维度 = 3,正则系数 = 0.1")
//    println(s"测试数据上的均方根误差（RMSE）= $rmse1")
//    println(s"测试数据上的平均绝对误差（MAE）= $mae1")
//    println(s"轮次 = 10,维度 = 3,正则系数 = 0.1")
//    println(s"测试数据上的均方根误差（RMSE）= $rmse2")
//    println(s"测试数据上的平均绝对误差（MAE）= $mae2")
//    println(s"轮次 = 10,维度 = 6,正则系数 = 0.1")
//    println(s"测试数据上的均方根误差（RMSE）= $rmse3")
//    println(s"测试数据上的平均绝对误差（MAE）= $mae3")
//    println(s"轮次 = 15,维度 = 3,正则系数 = 0.1")
//    println(s"测试数据上的均方根误差（RMSE）= $rmse4")
//    println(s"测试数据上的平均绝对误差（MAE）= $mae4")
//    println(s"轮次 = 15,维度 = 6,正则系数 = 0.1")
//    println(s"测试数据上的均方根误差（RMSE）= $rmse5")
//    println(s"测试数据上的平均绝对误差（MAE）= $mae5")
    println(s"轮次 = 15,维度 = 9,正则系数 = 0.1")
    println(s"测试数据上的均方根误差（RMSE）= $rmse6")
    println(s"测试数据上的平均绝对误差（MAE）= $mae6")

    // 创建 ALS 模型
//    val als = new ALS()
//      .setMaxIter(5) // 迭代次数，用于最小二乘交替迭代的次数
//      .setRank(3) // 隐向量的维度
//      .setRegParam(0.1) // 惩罚系数
//      .setUserCol("userId") // user_id
//      .setItemCol("itemId") // item_id
//      .setRatingCol("rating") // 评分列


    /***
     *  5,3,0.1 =
     * 测试数据上的均方根误差（RMSE）= 0.9800489334134103
     * 测试数据上的平均绝对误差（MAE）= 0.503805481573315
     *
     * 15,6,0.1
     * 测试数据上的均方根误差（RMSE）= 0.9592686818114753
     * 测试数据上的平均绝对误差（MAE）= 0.48160937365145484
     *
     */

//    // 训练模型
//    val model = als.fit(trainingData)
//
//    // 在测试集上评估模型
//
//    val predictions = model.transform(testData)
//      .withColumn("prediction", col("prediction").cast("double"))
//      .na.drop(Seq("prediction"))
//    // 展示测试集的预测结果
//    predictions.show(5)
//    println("--------预测值-------")
//
//    predictions.printSchema()
//
//
//    // 使用 evaluateModel 函数评估模型性能
//    val (rmse, mae) = evaluateModel(predictions)
//
//    // 打印结果
//    println(s"测试数据上的均方根误差（RMSE）= $rmse")
//    println(s"测试数据上的平均绝对误差（MAE）= $mae")




//    // 打印用户向量和物品向量
//    model.userFactors.show(truncate = false)
//    println("---------用户特征-----------")
//    model.itemFactors.show(truncate = false)
//    println("---------商品特征-----------")
//
//    println("-------开始推荐商品---------")
//    // 给所有用户推荐2个物品
//    //    model.recommendForAllUsers(2).show()
//    model.recommendForAllUsers(3).show(truncate = false)

    // 给所有用户推荐20个物品
    //    val recommendations = model.recommendForAllUsers(5).show()


    // 保存数据


    // 停止 SparkSession
    spark.stop()
  }
}
